﻿using BaseLibrary.CDataContext;
using BaseLibrary.CDataContext.CAttribute;
using BaseLibrary.ClassLibrary.EAFLog;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace DataSyn.MDL
{
    class Sync
    {
        class CSync_同步
        {
            public bool isRunSync { get; set; }
            /// <summary>
            /// 主表结构
            /// </summary>
            public DataTableattirbute tableSchema { get; set; }
            public string syncTableName { get; set; }
            public BaseLibrary.ClassLibrary.CThread thread = new BaseLibrary.ClassLibrary.CThread() { MaxThreadRunCount=1};
            public bool isRunning
            {
                get
                {
                    return thread.ActionCount > 0;
                }
            }
            int _pagerSize = 0;
            public int pageSize
            {
                get
                {
                    if (_pagerSize == 0)
                        if (tableSchema.ListColumns.Count(c => c.Att_DataType == System.Data.DbType.Binary) > 0)
                            _pagerSize = 10;
                        else
                            _pagerSize = 50;
                    return _pagerSize;
                }
            }
            /// <summary>
            /// 核查次数
            /// </summary>
            public int checkTimes { get; set; }
        }
        internal static string syncSchema = "sync";
        internal static string syncHead = "sync";
        internal static string syncTriggerHead = "syncTgr";
       BaseLibrary.ClassLibrary.CThread thread = new BaseLibrary.ClassLibrary.CThread();
        DAL.DB.Tables.TProject project = null;
        List<CSync_同步> list_Sync_同步 = new List<CSync_同步>();
        /// <summary>
        /// 重启
        /// </summary>
        public void ReStart()
        {
            StopSync();
            if (project.IsRunning)
                StartSync();
        }
        /// <summary>
        /// 同步表的同步结构
        /// </summary>
        DataTableattirbute syncTableSchema = new MDL.DataBase.SyncTableBase().GetSchema() as DataTableattirbute;
        List<DataTableattirbute> list_tableSchema = new List<DataTableattirbute>();
        public Sync(DAL.DB.Tables.TProject project)
        {
            this.project = project;
        }
        public void CreateMaster()
        {
            StopSync();
            var masterDC = MDL.CPublic.dc.GetMasterDC(project.ID);
            var clientDCs = MDL.CPublic.dc.GetClientDC(project.ID);
            if (masterDC == null)
                return;
            List<DataContext> list_sync_dc = new List<DataContext>();
            list_sync_dc.Add(masterDC);
            if (project.IsTowWay)
                list_sync_dc.AddRange(clientDCs);

            foreach (var dc in list_sync_dc)
            {
                ///创建同步架构
                if (dc.GetList<int>("select count(*) from sys.schemas  where name='{0}'", syncSchema).FirstOrDefault() == 0)
                {
                    dc.RunSql(string.Format("create schema {0}", syncSchema));
                }
                #region 删除不同步的触发器
                foreach (var triger in dc.GetTriggers().Where(c => c.TriggerName.IndexOf(syncTriggerHead) == 0))
                {
                    dc.RunSql(string.Format("drop trigger {0}", triger.TriggerName));
                }
                #endregion
                #region 删除不同步的表
                var syncTables = MDL.CPublic.dc.TDataTable.Where(c => c.ProjectID == project.ID).ToArray();
                foreach (var table in dc.GetTables().Where(c => c.SchemaName == syncSchema && syncTables.Count(d => d.TableName == c.TableName) > 0))
                {
                    dc.RunSql(string.Format("drop table {0}.{1}", table.SchemaName, table.TableName));
                }
                #endregion

            }

            foreach (var table in MDL.CPublic.dc.TDataTable.Where(c => c.ProjectID == project.ID).ToList())
                checkSyncTable(table.TableName);
            if (project.IsRunning)
                StartSync();
        }
        /// <summary>
        /// 取消初始化主数据库同步功能
        /// </summary>
        public void CancelMaster()
        {
            StopSync();
            project.IsRunning = false;
            foreach (var dc in MDL.CPublic.dc.GetAllDC(project.ID))
            {
                #region 删除同步的触发器
                var masf = dc.GetTriggers();
                foreach (var triger in dc.GetTriggers().Where(c => c.TriggerName.IndexOf(syncTriggerHead) == 0))
                {
                    dc.RunSql(string.Format("drop trigger {0}", triger.TriggerName));
                }
                #endregion
                #region 删除不同步的表
                var syncTables = MDL.CPublic.dc.TDataTable.Where(c => c.ProjectID == project.ID).ToArray();
                foreach (var table in dc.GetTables().Where(c => c.SchemaName == syncSchema))
                {
                    dc.RunSql(string.Format("drop table {0}.{1}", table.SchemaName, table.TableName));
                }
                #endregion
                ///删除同步架构
                if (dc.GetList<int>("select count(*) from sys.schemas  where name='{0}'", syncSchema).FirstOrDefault() > 0)
                {
                    dc.RunSql(string.Format("drop schema {0}", syncSchema));
                }
            }
        }
        /// <summary>
        /// 核查创建同步表
        /// </summary>
        /// <param name="tableName"></param>
        public void checkSyncTable(string tableName)
        {
            try
            {
                var tableSchema = GetTableSchema(tableName);
                var syncTableSchema = CreateSyncTableSchema(tableName);
                //创建主库同步表和触发器
                CreateDataBaseSync(MDL.CPublic.dc.GetMasterDC(project.ID), syncTableSchema, tableName);
                if (project.IsTowWay)
                    foreach (var client_dc in MDL.CPublic.dc.GetClientDC(project.ID))
                    {
                        client_dc.CheckTable(tableSchema);
                        CreateDataBaseSync(client_dc, syncTableSchema, tableName);
                    }
            }
            catch (Exception ex)
            {
                MDL.CPublic.log.Write(string.Format("核查创建同步表:{0}。{1}", tableName, ex.Message), MsgType.Error);
                MDL.CPublic.log.Write(ex);
            }
        }
        /// <summary>
        /// 清除同步表
        /// </summary>
        /// <param name="tableName"></param>
        public void cancelSyncTable(string tableName)
        {
            StopSync();
            foreach (var dc in MDL.CPublic.dc.GetAllDC(project.ID))
            {
                //删除触发器
                var List_Triggers = dc.GetTriggers().Where(c => c.TriggerName.IndexOf(syncTriggerHead) == 0).ToList();
                if (List_Triggers.Count(c => c.TriggerName == syncTriggerHead + tableName) > 0)
                    dc.RunSql(string.Format("drop trigger {0}", syncTriggerHead+tableName));
                //删除同步表
                if (dc.GetTables().Count(c => c.SchemaName == syncHead && c.TableName == (syncHead + tableName)) > 0)
                    dc.RunSql(string.Format("drop table {0}.{1}", syncHead, syncHead + tableName));
            }
            if (project.IsRunning)
                StartSync();
        }
        private string getDataBaseName(int id)
        {
            return MDL.CPublic.dc.TDataBase.Where(c => c.ID == id).ToList().Select(c =>string.Format("{0}:{1}",c.ConnectionID, c.DataBase)).FirstOrDefault();
        }
        /// <summary>
        /// 从主库得到表结构
        /// </summary>
        /// <param name="table_Name"></param>
        /// <returns></returns>
        private DataTableattirbute GetTableSchema(string table_Name)
        {
            return GetTableSchema(MDL.CPublic.dc.GetMasterDC(project.ID),table_Name);
        }
        /// <summary>
        /// 得到表结构
        /// </summary>
        /// <param name="table_Name"></param>
        /// <returns></returns>
        private DataTableattirbute GetTableSchema(DataContext dc,string table_Name)
        {
            var tableSchema = new DataTableattirbute() { Att_Name = table_Name };
            tableSchema.ListColumns.AddRange(dc.GetTableColumnsAttribute(tableSchema.Att_Name, tableSchema.Schema).ToList());
            return tableSchema;
        }
        /// <summary>
        ///创建同步表结构
        /// </summary>
        /// <param name="table_Name"></param>
        /// <returns></returns>
        public DataTableattirbute CreateSyncTableSchema(string table_Name)
        {
            var syncTableSchema = (new MDL.DataBase.SyncTableBase()).GetSchema() as DataTableattirbute;
            syncTableSchema.Schema = syncSchema;
            syncTableSchema.Att_Name = string.Format("{0}{1}", syncHead, table_Name);
            var pkcolumn = syncTableSchema.ListColumns.Select(c => c as DataColumnAttirbute).FirstOrDefault(c => c.Att_ISPrimary);
            if (pkcolumn == null)
                pkcolumn.Att_PrimaryKeyName = pkcolumn.Att_DFName = string.Format("PF{0}{1}", syncSchema, table_Name);
            else
                pkcolumn.Att_PrimaryKeyName = "PK_" + syncSchema + table_Name;
            foreach (var columns_Schema in GetTableSchema(table_Name).ListColumns.Cast<DataColumnAttirbute>())
            {
                columns_Schema.Att_AutoIncrement = false;
                columns_Schema.Att_ISPrimary = false;
                syncTableSchema.ListColumns.Add(columns_Schema);
            }
            return syncTableSchema;
        }
        /// <summary>
        /// 创建同步表及触发器
        /// </summary>
        /// <param name="dc"></param>
        /// <param name="syncTableSchema"></param>
        /// <param name="table_tem"></param>
        public void CreateDataBaseSync(DataContext dc,DataTableattirbute syncTableSchema, string table_tem)
        {
            var tableName = table_tem;
            #region 同步表
            dc.CheckTable(new DataTableattirbute[] { syncTableSchema });
            #endregion
            #region 触发器
            var List_Triggers = dc.GetTriggers().Where(c => c.TriggerName.IndexOf(syncTriggerHead) == 0).ToList();
            var sql_Builder = new StringBuilder();
            if (List_Triggers.Count(c => c.TriggerName == string.Format("{0}{1}", syncTriggerHead, tableName)) == 0)
            {
                sql_Builder.AppendLine(string.Format("create trigger {0}{1} on {1} for insert,update,delete as ", syncTriggerHead, tableName));
            }
            else
            {
                sql_Builder.AppendLine(string.Format("alter trigger {0}{1} on {1} for insert,update,delete as ",  syncTriggerHead, tableName));
            }
            //判断是否删除
            sql_Builder.AppendLine("begin try");
            sql_Builder.AppendLine("if exists (select 1 from deleted) and  exists (select 1 from inserted)");
            sql_Builder.AppendLine("begin");
            sql_Builder.AppendLine(string.Format("insert into {0}.{1}({2}) select 'update',getdate(),{3} from inserted", syncSchema, syncTableSchema.Att_Name, string.Join(",", syncTableSchema.ListColumns.Skip(1).Select(c => c.Att_Column).ToArray()), string.Join(",", syncTableSchema.ListColumns.Skip(3).Select(c => c.Att_Column).ToArray())));
            sql_Builder.AppendLine("end");
            //判断是否删除
            sql_Builder.AppendLine("else if exists (select 1 from deleted)");
            sql_Builder.AppendLine("begin");
            sql_Builder.AppendLine(string.Format("insert into {0}.{1}({2}) select 'delete',getdate(),{3} from deleted", syncSchema, syncTableSchema.Att_Name, string.Join(",", syncTableSchema.ListColumns.Skip(1).Select(c => c.Att_Column).ToArray()), string.Join(",", syncTableSchema.ListColumns.Skip(3).Select(c => c.Att_Column).ToArray())));
            sql_Builder.AppendLine("end");
            //判断是否插入
            sql_Builder.AppendLine("else if exists (select 1 from inserted)");
            sql_Builder.AppendLine("begin");
            sql_Builder.AppendLine(string.Format("insert into {0}.{1}({2}) select 'insert',getdate(),{3} from inserted", syncSchema, syncTableSchema.Att_Name, string.Join(",", syncTableSchema.ListColumns.Skip(1).Select(c => c.Att_Column).ToArray()), string.Join(",", syncTableSchema.ListColumns.Skip(3).Select(c => c.Att_Column).ToArray())));
            sql_Builder.AppendLine("end");
            sql_Builder.AppendLine("end try");
            sql_Builder.AppendLine("begin catch");
            sql_Builder.AppendLine("end catch");
            dc.RunSql(sql_Builder.ToString());
            #endregion
        }
        public void StartSync()
        {
            thread.Start(() => {
                MDL.CPublic.dc.TSyncProcess.Clear();
                while (true)
                {
                    foreach (var table in MDL.CPublic.dc.TDataTable.Where(c => c.ProjectID == project.ID).ToArray())
                    {
                        CheckTableData(table);
                    }
                }
            });
        }
        private void CheckTableData(DAL.DB.Tables.TDataTable table)
        {
            #region 同步表处理
            foreach (var dataBaseFrom in MDL.CPublic.dc.TDataBase.Where(c => c.ProjectID == project.ID).ToArray().OrderByDescending(c => c.IsMasterDB))
            {
                if (!project.IsTowWay && !dataBaseFrom.IsMasterDB) continue;
                var dcSource = MDL.CPublic.dc.GetDataBaseDC(dataBaseFrom.ID);
                try
                {
                    if (!dcSource.ConnTest)
                        throw new Exception(string.Format("{0},数据库连接失败", dcSource.ConnectionString));
                    var sync = list_Sync_同步.FirstOrDefault(c => c.tableSchema.Att_Name == table.TableName);
                    if (sync == null)
                        list_Sync_同步.Add(sync = new CSync_同步() { tableSchema = GetTableSchema(dcSource, table.TableName), syncTableName = syncSchema + "." + syncHead + table.TableName });
                    if (sync.isRunSync && sync.isRunning) return;
                    var syncDataCount = dcSource.GetList<int>("select count(*) from " + sync.syncTableName).FirstOrDefault();
                    sync.checkTimes++;
                    if (syncDataCount > 0)
                    {
                        sync.isRunSync = true;
                        sync.thread.Start(() =>
                        {
                            List<int> ErrConnDataBaseID = new List<int>();
                            try
                            {
                                foreach (var dataBaseTo in MDL.CPublic.dc.TDataBase.Where(c => c.ProjectID == project.ID && c.ID != dataBaseFrom.ID).ToList())
                                {
                                    var dcClient = MDL.CPublic.dc.GetDataBaseDC(dataBaseTo.ID);
                                    try
                                    {
                                        if (!dcClient.ConnTest)
                                        {
                                            ErrConnDataBaseID.Add(dataBaseTo.ID);
                                            throw new Exception(string.Format("{0},数据库连接失败", dcClient.ConnectionString));
                                        }
                                        // MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理同步{2}表……", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), table.TableName));
                                        int checkCount = 0;
                                        var lastID =  BaseLibrary.ClassLibrary.DataConver.ToInt(MDL.CPublic.dc.TSyncProcess.Where(c => c.DataBaseSourceID == dataBaseFrom.ID && c.DataBaseToID == dataBaseTo.ID && c.TableID == table.ID).Max(c => c.LastSyncID));
                                        BaseLibrary.ClassLibrary.PagerBack<Dictionary<string, object>> pager = new BaseLibrary.ClassLibrary.PagerBack<Dictionary<string, object>>();
                                        do
                                        {
                                            try
                                            {
                                                pager = dcSource.PagerByModule(string.Format("select * from " + sync.syncTableName + " where DataSyncID >" + lastID.ToString()), " order by DataSyncID", pager.PageIndex + 1, sync.pageSize);
                                                if (pager.Rows.Count == 0) break;
                                                Reader同步表(table.TableName, pager.Rows.ToList(), dataBaseFrom.ID, dataBaseTo.ID, sync.tableSchema);
                                                MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理同步{2}表数据{3}条第{4}/{5}页", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name, pager.Rows.Count, pager.PageIndex, (int)Math.Ceiling((decimal)pager.TotalCount / pager.PageSize)));
                                                checkCount = 0;
                                            }
                                            catch (Exception ex)
                                            {
                                                if (ex.Message.IndexOf("找不到存储过程 'ModulePager'。") > 0)
                                                    dcSource.CheckModule(new ModulePager());
                                                if (ex.Message.IndexOf(string.Format("对象名 '{0}' 无效。", sync.syncTableName)) > 0)
                                                    checkSyncTable(sync.tableSchema.Att_Name);
                                                MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理同步{2}表出现异常：{3}", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), table.TableName, ex.Message));
                                                MDL.CPublic.log.Write(ex);
                                                checkCount++;
                                                System.Threading.Thread.Sleep(1000);
                                            }
                                        }
                                        while (checkCount < 3);
                                        if (checkCount == 3)
                                        {
                                            ErrConnDataBaseID.Add(dataBaseTo.ID);
                                            MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理同步{2}表异常。退出同步表同步", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name), MsgType.Error);
                                        }
                                        else
                                            MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}处理同步{2}表完成。", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), table.TableName));

                                    }
                                    catch (Exception ex)
                                    {
                                        MDL.CPublic.log.Write(ex);
                                    }
                                }
                                var minSyncID = MDL.CPublic.dc.GetList<int>(string.Format(@"select min(ifnull(TSyncProcess.LastSyncID,0)) as maxSyncID from TProject
left join TDataBase on TDataBase.ProjectID=TProject.ID
left JOIN TDataTable on TDataTable.ProjectID=TProject.ID
left join TSyncProcess on TSyncProcess.DataBaseToID=TDataBase.ID and TSyncProcess.TableID=TDataTable.ID
where TableName='{0}' and TSyncProcess.DataBaseSourceID={1}", table.TableName, dataBaseFrom.ID) + (ErrConnDataBaseID.Count > 0 ? " and TSyncProcess.DataBaseToID not in (" + string.Join(",", ErrConnDataBaseID.Select(c => c.ToString()).ToArray()) + ")" : "")).FirstOrDefault();
                                if (minSyncID > 0)
                                    dcSource.RunSql(string.Format("delete from {0} where DataSyncID<={1}", sync.syncTableName, minSyncID));
                            }
                            catch (Exception ex)
                            { }
                            sync.isRunSync = false;
                        });
                        return;
                    }
                    System.Threading.Thread.Sleep(1000);
                }
                catch (Exception ex)
                {
                    if (ex.Message.IndexOf(string.Format("对象名 '{0}' 无效。", table.TableName)) > 0
                        || ex.Message.IndexOf(string.Format("对象名 '{0}' 无效。", (syncSchema + "." + syncHead + table.TableName))) > 0)
                        checkSyncTable(table.TableName);
                    MDL.CPublic.log.Write(ex);
                }
            }
            #endregion
            #region 同步表没有数据时从主表检测
            var databases = MDL.CPublic.dc.TDataBase.Where(c => c.ProjectID == project.ID).ToArray().OrderByDescending(c => c.IsMasterDB);
            foreach (var dataBaseFrom in databases)
            {
                if (!project.IsTowWay && !dataBaseFrom.IsMasterDB) continue;
                var dcSource = MDL.CPublic.dc.GetDataBaseDC(dataBaseFrom.ID);
                try
                {
                    if (!dcSource.ConnTest)
                        throw new Exception(string.Format("{0},数据库连接失败", dcSource.ConnectionString));
                    var sync = list_Sync_同步.FirstOrDefault(c => c.tableSchema.Att_Name == table.TableName);
                    if (sync == null)
                        list_Sync_同步.Add(sync = new CSync_同步() { tableSchema = GetTableSchema(dcSource, table.TableName), syncTableName = syncSchema + "." + syncHead + table.TableName });
                    if (sync.isRunning) continue;
                    if (sync.checkTimes < 10) continue;
                    sync.checkTimes = 0;

                    foreach (var dataBaseTo in MDL.CPublic.dc.TDataBase.Where(c => c.ProjectID == project.ID && c.ID != dataBaseFrom.ID).ToList())
                    {
                        var dcClient = MDL.CPublic.dc.GetDataBaseDC(dataBaseTo.ID);
                        try
                        {
                            if (!dcClient.ConnTest)
                                throw new Exception(string.Format("{0},数据库连接失败", dcClient.ConnectionString));
                            BaseLibrary.ClassLibrary.PagerBack<Dictionary<string, object>> pager = new BaseLibrary.ClassLibrary.PagerBack<Dictionary<string, object>>();
                            var column_auto = sync.tableSchema.ListColumns.Cast<DataColumnAttirbute>().FirstOrDefault(c => c.Att_AutoIncrement);
                            if (column_auto == null)
                            {
                                var clientCount = dcClient.GetList<int>("select count(*) as clientCount from " + sync.tableSchema.Att_Name).FirstOrDefault();
                                var sourcCount = dcSource.GetList<int>("select (select count(*) from " + sync.tableSchema.Att_Name + ") as sourceCount").FirstOrDefault();
                                if (clientCount >= sourcCount)
                                    continue;
                                sync.thread.Start(() =>
                                {
                                    System.Threading.Thread.Sleep(10000);
                                    int checkCount = 0;
                                    do
                                    {
                                        try
                                        {
                                            pager = dcSource.PagerByModule("select * from " + sync.tableSchema.Att_Name, " order by " + string.Join(",", sync.tableSchema.ListColumns.Select(c => c.Att_DataSourceColumnName).ToArray()), pager.PageIndex + 1, sync.pageSize);
                                            if (pager.Rows.Count == 0) break;
                                            Reader同步表(sync.tableSchema.Att_Name, pager.Rows.ToList(), dataBaseFrom.ID, dataBaseTo.ID, sync.tableSchema);
                                            MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理主{2}表数据{3}条第{4}/{5}页", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name, pager.Rows.Count, pager.PageIndex, (int)Math.Ceiling((decimal)pager.TotalCount / pager.PageSize)));
                                            checkCount = 0;
                                        }
                                        catch (Exception ex)
                                        {
                                            if (ex.Message.IndexOf("找不到存储过程 'ModulePager'。") > 0)
                                                dcSource.CheckModule(new ModulePager());
                                            if (ex.Message.IndexOf(string.Format("对象名 '{0}' 无效。", sync.tableSchema.Att_Name)) > 0)
                                                checkSyncTable(sync.tableSchema.Att_Name);
                                            MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理主{2}表异常:{3}", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name, ex.Message), MsgType.Error);
                                            MDL.CPublic.log.Write(ex);
                                            pager.PageIndex = Math.Max(1, pager.PageIndex - 1);
                                           System.Threading.Thread.Sleep(1000);
                                            checkCount++;
                                        }
                                    }
                                    while (checkCount < 3);
                                    if (checkCount == 3)
                                        MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理主{2}表异常。退出主表同步", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name),MsgType.Error);
                                    else
                                        MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}处理主{2}表完成。", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), table.TableName));
                                });
                            }
                            else
                            {
                                var clientValues = dcClient.GetList("select count(*) as clientCount,max(" + column_auto.Att_Column + ") as maxID from " + sync.tableSchema.Att_Name).FirstOrDefault();
                                var clientCount = BaseLibrary.ClassLibrary.DataConver.ToInt(clientValues["clientCount"]);
                                var clientMaxID =  BaseLibrary.ClassLibrary.DataConver.ToInt(clientValues["maxID"]);
                                var where = " where " + column_auto.Att_Column + " >" + clientMaxID.ToString();
                                var sourceValues = dcSource.GetList("select (select count(*) from " + sync.tableSchema.Att_Name + ") as sourceCount,(select count(*) from " + sync.tableSchema.Att_Name + where + ") as diffCount").FirstOrDefault();
                                var diffCount =  BaseLibrary.ClassLibrary.DataConver.ToInt(sourceValues["diffCount"]);
                                var sourcCount =  BaseLibrary.ClassLibrary.DataConver.ToInt(sourceValues["sourceCount"]);
                                if (diffCount == 0)
                                {
                                    if (clientCount >= sourcCount)
                                        continue;
                                    else
                                        where = "";
                                }
                                sync.thread.Start(() =>
                                {
                                    System.Threading.Thread.Sleep(10000);
                                    int checkCount = 0;
                                    do
                                    {
                                        try
                                        {
                                            var str = string.Format("select * from " + sync.tableSchema.Att_Name + where);
                                            pager = dcSource.PagerByModule("select * from " + sync.tableSchema.Att_Name + where, " order by " + column_auto.Att_Column, pager.PageIndex + 1, sync.pageSize);
                                            if (pager.Rows.Count == 0) break;
                                            Reader同步表(sync.tableSchema.Att_Name, pager.Rows.ToList(), dataBaseFrom.ID, dataBaseTo.ID, sync.tableSchema);
                                            MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理主{2}表数据{3}条第{4}/{5}页", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name, pager.Rows.Count, pager.PageIndex, (int)Math.Ceiling((decimal)pager.TotalCount / pager.PageSize)));
                                            checkCount = 0;
                                        }
                                        catch (Exception ex)
                                        {
                                            if (ex.Message.IndexOf("找不到存储过程 'ModulePager'。") > 0)
                                                dcSource.CheckModule(new ModulePager());
                                            if (ex.Message.IndexOf(string.Format("对象名 '{0}' 无效。", sync.tableSchema.Att_Name)) > 0)
                                                checkSyncTable(sync.tableSchema.Att_Name);
                                            MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理主{2}表异常:{3}", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name, ex.Message),MsgType.Error);
                                            MDL.CPublic.log.Write(ex);
                                            pager.PageIndex = Math.Max(1, pager.PageIndex - 1);
                                            checkCount++;
                                            System.Threading.Thread.Sleep(1000);
                                        }
                                    }
                                    while (checkCount < 3);
                                    if (checkCount == 3)
                                        MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}正在处理主{2}表异常。退出主表同步", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), sync.tableSchema.Att_Name), MsgType.Error);
                                    else
                                        MDL.CPublic.log.Write(string.Format("当前从dataBase:{0}到dataBase:{1}处理主{2}表完成。", getDataBaseName(dataBaseFrom.ID), getDataBaseName(dataBaseTo.ID), table.TableName));
                                });
                            }
                            break;
                        }
                        catch (Exception ex)
                        {
                            MDL.CPublic.log.Write(ex.Message, MsgType.Error);
                        }
                    }
                }
                catch (Exception ex)
                { }
            }
            #endregion
        }
        public void StopSync()
        {
            foreach (var sync in list_Sync_同步)
                sync.thread.Close();
            thread.Close();
        }
        private void Reader同步表(string tableName,List<Dictionary<string, object>> list_values, int dbSourceID, int dbToID,  DataTableattirbute tableSchema )
        {
            Writer(list_values, dbSourceID, dbToID, tableName, tableSchema);
        }
        /// <summary>
        /// 向其他库写入数据
        /// </summary>
        /// <param name="list_rows"></param>
        /// <param name="clientDataBase"></param>
        /// <param name="tableName"></param>
        /// <param name="tableSchema"></param>
        /// <param name="syncTableSchema"></param>
        /// <param name="list_SyncRows"></param>
        private void Writer(List<Dictionary<string, object>> list_rows, int dbSourceID, int dbToID, string tableName, DataTableattirbute tableSchema)
        {
            if (tableName.ToLower() == "tdataphoto")
            {
            }

            DataContext sourceDC = MDL.CPublic.dc.GetDataBaseDC(dbSourceID);
            DataContext clientDC = MDL.CPublic.dc.GetDataBaseDC(dbToID);
            int lastSyncID = 0;
            var columnAuto = tableSchema.ListColumns.Cast<DataColumnAttirbute>().FirstOrDefault(c => c.Att_AutoIncrement);
            if (columnAuto != null)
                clientDC.RunSql("SET IDENTITY_INSERT " + tableName + " OFF;");
            using (var tra = new DBTransaction(clientDC.OpenConn()))
            { 
                if (columnAuto != null)
                {
                    
                    tra.CreateRunCommandCreateCommand("SET IDENTITY_INSERT " + tableName + " ON;");
                }
                #region 关闭触发器
                var triggerName = syncTriggerHead + tableName;
                tra.CreateRunCommandCreateCommand(string.Format(@"if exists(select name from sys.triggers where name='{0}')
ALTER TABLE {1} DISABLE TRIGGER {0}", triggerName, tableName));
                #endregion
                #region 数据同步
                for (int irow = 0; irow < list_rows.Count; irow++)
                {
                    StringBuilder sb_sql = new StringBuilder();
                    var row = list_rows[irow];
                    List<System.Data.IDbDataParameter> list_values = new List<System.Data.IDbDataParameter>();
                    foreach (KeyValuePair<string, object> column_Value in row)
                    {
                        var col = tableSchema.ListColumns.Select(c => c as DataColumnAttirbute).FirstOrDefault(c => c.Att_DataSourceColumnName == column_Value.Key);
                        if (col == null) continue;
                        list_values.Add(clientDC.GetParameter(list_values.Count, column_Value.Value, col));
                    }
                    if (row.ContainsKey("DataSyncID") && row.ContainsKey("DataSyncType"))
                    {
                        lastSyncID = Math.Max(lastSyncID, BaseLibrary.ClassLibrary.DataConver.ToInt(row["DataSyncID"]));
                        if (string.Format("{0}", row["DataSyncType"]).ToLower() == "delete")
                        {
                            if (columnAuto == null)
                            {
                                #region 判断源主表不存在才删除
                                using (var conn = sourceDC.OpenConn())
                                using (var comm = conn.CreateCommand("select count(*) from "
                                  + tableName
                                  + " where " + string.Join(" and ", list_values.Select(c => string.Format("{0}={1}", c.SourceColumn, c.ParameterName)).ToArray()), list_values.ToArray()))
                                {
                                    if (BaseLibrary.ClassLibrary.DataConver.ToInt(comm.ExecuteScalar()) == 0)
                                    {
                                        //开始处理
                                        sb_sql.AppendLine("delete from "
                                          + tableName
                                          + " where " + string.Join(" and ", list_values.Select(c => string.Format("{0}={1}", c.SourceColumn, c.ParameterName)).ToArray()));
                                        tra.CreateRunCommandCreateCommand(sb_sql.ToString(), list_values.ToArray());
                                    }
                                }
                                #endregion
                            }
                            else
                            {
                                using (var conn = sourceDC.OpenConn())
                                using (var comm = conn.CreateCommand("select count(*) from "
                                       + tableName
                                       + " where " + list_values.Where(c => c.SourceColumn == columnAuto.Att_DataSourceColumnName).Select(c => string.Format("{0}={1};", c.SourceColumn, c.ParameterName)).FirstOrDefault(), list_values.Where(c => c.SourceColumn == columnAuto.Att_DataSourceColumnName).ToArray()))
                                {
                                    if (BaseLibrary.ClassLibrary.DataConver.ToInt(comm.ExecuteScalar()) == 0)
                                    {
                                        sb_sql.AppendLine("delete from "
                                      + tableName
                                      + " where " + list_values.Where(c => c.SourceColumn == columnAuto.Att_DataSourceColumnName).Select(c => string.Format("{0}={1};", c.SourceColumn, c.ParameterName)).FirstOrDefault());
                                        tra.CreateRunCommandCreateCommand(sb_sql.ToString(), list_values.Where(c => c.SourceColumn == columnAuto.Att_DataSourceColumnName).ToArray());
                                    }
                                }
                            }
                            continue;
                        }
                    }
                    if (columnAuto == null)
                    {
                        sb_sql.AppendLine("if((select count(*) from " + tableName + " where " + string.Join(" and ", list_values.Select(c => string.Format("{0}={1}", c.SourceColumn, c.ParameterName)).ToArray()) + ")=0)");
                        sb_sql.AppendLine("Begin");
                        sb_sql.AppendLine("insert into " + tableName + " (" + string.Join(",", list_values.Select(c => c.SourceColumn).ToArray()) + ") values(" + string.Join(",", list_values.Select(c => c.ParameterName).ToArray()) + ");");
                        sb_sql.AppendLine("End");
                    }
                    else
                    {
                        sb_sql.AppendLine("if((select count(*) from " + tableName + " where " + list_values.Where(c => c.SourceColumn == columnAuto.Att_DataSourceColumnName).Select(c => string.Format("{0}={1}", c.SourceColumn, c.ParameterName)).FirstOrDefault() + ")=0)");
                        sb_sql.AppendLine("Begin");
                        sb_sql.AppendLine("insert into " + tableName + " (" + string.Join(",", list_values.Select(c => c.SourceColumn).ToArray()) + ") values(" + string.Join(",", list_values.Select(c => c.ParameterName).ToArray()) + ");");
                        sb_sql.AppendLine("End");
                        sb_sql.AppendLine("Else");
                        sb_sql.AppendLine("Begin");
                        sb_sql.AppendLine("update " + tableName + " set " + string.Join(",", list_values.Where(c => c.SourceColumn != columnAuto.Att_DataSourceColumnName).Select(c => string.Format("{0}={1}", c.SourceColumn, c.ParameterName)).ToArray()) + " where " + string.Join(",", list_values.Where(c => c.SourceColumn == columnAuto.Att_DataSourceColumnName).Select(c => string.Format("{0}={1}", c.SourceColumn, c.ParameterName)).ToArray()) + ";");
                        sb_sql.AppendLine("End");
                    }
                    tra.CreateRunCommandCreateCommand(sb_sql.ToString(), list_values.ToArray());
                }
                #endregion
                if (columnAuto != null)
                    tra.CreateRunCommandCreateCommand("SET IDENTITY_INSERT " + tableName + " OFF;");
                //重新打开触发器
                #region 打开触发器
                tra.CreateRunCommandCreateCommand(string.Format(@"if exists(select name from sys.triggers where name='{0}')
ALTER TABLE {1} ENABLE TRIGGER {0}", triggerName, tableName));
                #endregion
                tra.Commit();
                tra.DBConn.Dispose();
            }
            ///同步后更新记录同步进度
            if (lastSyncID > 0)
            {
                 BaseLibrary.ClassLibrary.Win32.WinForm.Invoke(() =>
                {
                    var table = MDL.CPublic.dc.TDataTable.FirstOrDefault(c => c.ProjectID == project.ID && c.TableName == tableName);
                    if (table != null)
                    {
                        var pro = MDL.CPublic.dc.TSyncProcess.FirstOrDefault(c => c.DataBaseSourceID == dbSourceID && c.DataBaseToID == dbToID && c.TableID == table.ID);
                        if (pro == null)
                            pro = new DAL.DB.Tables.TSyncProcess() { DataBaseSourceID = dbSourceID, DataBaseToID = dbToID, TableID = table.ID };
                        pro.LastSyncID = lastSyncID;
                        MDL.CPublic.dc.Save(pro);
                    }
                });
            }
        }
    }
}
